package net.bwie.flink;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Flink CDC 实时采集MySQL表数据：全量 + 增量
 * @author xuanyu
 * @date 2025/10/20
 */
public class _01CdcMysqlDemo {

	public static void main(String[] args) throws Exception{
		//1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		//2. 数据源-source
		/*
		Flink CDC 属于Flink Source数据源，实时从数据库表中获取数据
			核心参数：启动模式startup mode
				1. 全量采集数据
					先读取表中历史数据，后监控日志获取增量数据
				2. 增量采集数据
					直接监控日志获取增量数据

				【增量数据：插入insert新数据、更新update老新数据、删除delete过期数据】

		 */
		env.enableCheckpointing(3000);
		MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
			.hostname("node101")
			.port(3306)
			.databaseList("flink_mall")
			.tableList("flink_mall.comment_info")
			.username("root")
			.password("123456")
			.startupOptions(StartupOptions.initial())  // 全量
//			.startupOptions(StartupOptions.latest()) // 增量
			.deserializer(new JsonDebeziumDeserializationSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source"
		);
		stream.print("cdc");

		//5.触发执行
		env.execute("CdcMysqlDemo") ;
	}

}
